package com.larry.spark.test

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}


/**
 * 编号	字段名称	字段类型	字段含义
1	date	String	用户点击行为的日期
2	user_id	Long	用户的ID
3	session_id	String	Session的ID
4	page_id	Long	某个页面的ID
5	action_time	String	动作的时间点
6	search_keyword	String	用户搜索的关键词
7	click_category_id	Long	某一个商品品类的ID
8	click_product_id	Long	某一个商品的ID
9	order_category_ids	String	一次订单中所有品类的ID集合
10	order_product_ids	String	一次订单中所有商品的ID集合
11	pay_category_ids	String	一次支付中所有品类的ID集合
12	pay_product_ids	String	一次支付中所有商品的ID集合
13	city_id	Long	城市 id

 */
object Top10 {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("rdd")
    val sc = new SparkContext(conf)

    val dataLine = sc.textFile("data/user_visit_action.txt")

    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //点击，下单，支付
    //点击数据
    val clickData: RDD[(String, Int)] = dataLine.filter(
      line => {
        val lines = line.split("_")
        val click = lines(6)
        click != "-1"
      }
    ).map(
      line => {
        val lines = line.split("_")
        val click = lines(6)
        (click, 1)
      }
    ).reduceByKey(_ + _)

    //下单数据
    val orderData = dataLine.filter(
      line => {
        val lines = line.split("_")
        val order = lines(8)
        order != "null"
      }
    ).flatMap(
      line => {
        val lines = line.split("_")
        val order = lines(8)
        val orders = order.split(",")
        orders.map((_,1))
      }
    ).reduceByKey(_+_)

    //支付数据
    val payData = dataLine.filter(
      line => {
        val lines = line.split("_")
        val order = lines(10)
        order != "null"
      }
    ).flatMap(
      line => {
        val lines = line.split("_")
        val order = lines(10)
        val orders = order.split(",")
        orders.map((_,1))
      }
    ).reduceByKey(_+_)

    //聚合变为tuple后综合排序
    val top = clickData.cogroup(orderData, payData).map{
      case (id,(click,order,pay)) => {
        var clickCount = 0
        var orderCount = 0
        var payCount = 0

        val it1 = click.iterator
        if (it1.hasNext) {
          clickCount = it1.next()
        }

        val it2 = order.iterator
        if (it2.hasNext) {
          orderCount = it2.next()
        }

        val it3 = pay.iterator
        if (it3.hasNext) {
          payCount = it3.next()
        }

        (id,(clickCount,orderCount,payCount))
      }
    }.sortBy(_._2,false).take(10)   //降序取前10


    top.foreach(println)

    sc.stop()
  }

}
